package com.dataconver.sync;

import com.dataconver.config.ConfigManager;
import com.dataconver.db.DatabaseManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.HashSet;
import java.util.Set;

/**
 * UPSERT同步策略
 * 直接INSERT源表数据到目标表，让数据库的UNIQUE约束处理重复数据
 * 保护目标表中的历史数据不被删除
 */
public class UpsertSyncStrategy implements SyncStrategy {
    private static final Logger logger = LoggerFactory.getLogger(UpsertSyncStrategy.class);
    
    private final String sourceTable;        // 源表名
    private final String keyField;           // 关键字段（IMEI）
    private final String targetTable;        // 目标表名
    private final int batchSize;             // 批处理大小
    private final Map<String, String> columnMapping; // 列名映射
    
    public UpsertSyncStrategy(String sourceTable, String keyField, String targetTable) {
        this.sourceTable = sourceTable;
        this.keyField = keyField;
        this.targetTable = targetTable;
        this.batchSize = ConfigManager.getInt("sync.batchSize", 1000);
        this.columnMapping = loadColumnMapping();
    }
    
    /**
     * 加载列名映射配置
     */
    private Map<String, String> loadColumnMapping() {
        Map<String, String> mapping = new HashMap<>();
        
        try {
            String tableConfigKey = "sync.table." + targetTable + ".columnMapping";
            Map<String, Object> map = ConfigManager.getMap(tableConfigKey);
            if (map != null) {
                for (Map.Entry<String, Object> e : map.entrySet()) {
                    if (e.getKey() != null && e.getValue() != null) {
                        mapping.put(e.getKey(), String.valueOf(e.getValue()));
                    }
                }
            }
        } catch (Exception ex) {
            logger.warn("读取列名映射失败，使用默认映射", ex);
        }
        return mapping;
    }
    
    @Override
    public SyncResult execute(String tableName) {
        logger.info("开始执行UPSERT同步策略");
        logger.info("源表: {}, 目标表: {}, 关键字段: {}", sourceTable, tableName, keyField);
        logger.info("同步策略: 全量同步，按{}字段进行UPSERT操作，保护目标表历史数据", keyField);
        
        SyncResult result = new SyncResult();
        result.setTableName(tableName);
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 获取源表列
            logger.info("正在获取源表 {} 的列信息...", sourceTable);
            List<String> columns = getSourceTableColumns();
            logger.info("源表 {} 共有 {} 列", sourceTable, columns.size());

            // 确保目标表存在
            logger.info("正在检查目标表 {} 是否存在...", tableName);
            ensureTargetTableExists(tableName, columns);

            // 在同步前，确保目标表的关键字段建立唯一索引，并清理重复数据
            logger.info("正在确保目标表 {} 的关键字段 {} 具有唯一索引...", tableName, keyField);
            ensureTargetKeyUniqueness(tableName);

            // 执行流式UPSERT同步
            logger.info("开始执行流式UPSERT同步...");
            UpsertSyncStats stats = executeStreamingUpsert(tableName, columns, result);
            
            result.setTotalRows(stats.totalRows);
            result.setUpdatedRows(stats.updatedRows);
            result.setInsertedRows(stats.insertedRows);
            result.setDeletedRows(stats.deletedRows);
            result.setSuccess(true);
            result.setEndTime(System.currentTimeMillis());
            
            // 注释掉啰嗦的日志，只保留同步完成统计
            // logger.info("表 {} UPSERT同步完成，共同步 {} 行数据", tableName, stats.totalRows);
            
        } catch (Exception e) {
            logger.error("UPSERT同步表 {} 失败", tableName, e);
            result.setSuccess(false);
            result.setErrorMessage(e.getMessage());
            result.setEndTime(System.currentTimeMillis());
        }
        
        return result;
    }
    
    /**
     * 获取源表列信息
     */
    private List<String> getSourceTableColumns() {
        List<String> columns = new ArrayList<>();
        
        try (Connection conn = DatabaseManager.getSqlserverConnection()) {
            DatabaseMetaData metaData = conn.getMetaData();
            ResultSet rs = metaData.getColumns(null, null, sourceTable, null);
            
            while (rs.next()) {
                columns.add(rs.getString("COLUMN_NAME"));
            }
            
        } catch (SQLException e) {
            logger.error("获取源表列信息失败: {}", sourceTable, e);
        }
        
        return columns;
    }
    
    /**
     * 确保目标表存在
     */
    private void ensureTargetTableExists(String tableName, List<String> columns) throws SQLException {
        try (Connection conn = DatabaseManager.getMysqlConnection()) {
            DatabaseMetaData metaData = conn.getMetaData();
            ResultSet tables = metaData.getTables(null, null, tableName, new String[]{"TABLE"});
            
            if (!tables.next()) {
                createTargetTable(tableName, columns, conn);
            }
        }
    }
    
    /**
     * 创建目标表
     */
    private void createTargetTable(String tableName, List<String> columns, Connection conn) throws SQLException {
        StringBuilder createSql = new StringBuilder();
        createSql.append("CREATE TABLE IF NOT EXISTS `").append(tableName).append("` (");
        
        for (int i = 0; i < columns.size(); i++) {
            if (i > 0) createSql.append(", ");
            String targetColumn = columnMapping.getOrDefault(columns.get(i), columns.get(i));
            createSql.append("`").append(targetColumn).append("` VARCHAR(255)");
        }
        
        // 将key字段设置为唯一索引以支持ON DUPLICATE KEY
        createSql.append(", UNIQUE KEY `uk_").append(keyField.toLowerCase()).append("` (`").append(columnMapping.getOrDefault(keyField, keyField)).append("`)");
        createSql.append(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
        
        try (Statement stmt = conn.createStatement()) {
            stmt.execute(createSql.toString());

        }
    }

    /**
     * 构建MySQL UPSERT PreparedStatement SQL
     */
    private String buildUpsertSql(String tableName, List<String> columns) {
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO `").append(tableName).append("` (");
        for (int i = 0; i < columns.size(); i++) {
            if (i > 0) sql.append(", ");
            String targetColumn = columnMapping.getOrDefault(columns.get(i), columns.get(i));
            sql.append("`").append(targetColumn).append("`");
        }
        sql.append(") VALUES (");
        for (int i = 0; i < columns.size(); i++) {
            if (i > 0) sql.append(", ");
            sql.append("?");
        }
        sql.append(") ON DUPLICATE KEY UPDATE ");
        for (int i = 0; i < columns.size(); i++) {
            if (i > 0) sql.append(", ");
            String targetColumn = columnMapping.getOrDefault(columns.get(i), columns.get(i));
            sql.append("`").append(targetColumn).append("` = VALUES(`").append(targetColumn).append("`)");
        }
        return sql.toString();
    }

    /**
     * 流式读取SQLServer并直接批写MySQL
     */
    private UpsertSyncStats executeStreamingUpsert(String tableName, List<String> columns, SyncResult result) throws SQLException {
        UpsertSyncStats stats = new UpsertSyncStats();
        stats.totalRows = 0;
        stats.updatedRows = 0;
        stats.insertedRows = 0;
        stats.deletedRows = 0;

        // 1) 先删除目标表中源表没有的记录（同步删除）
        // 注释掉删除逻辑，保护目标表中的历史数据不被误删
        // stats.deletedRows = deleteOrphanedRecords(tableName);
        // logger.info("删除了 {} 条目标表中源表没有的记录", stats.deletedRows);
        stats.deletedRows = 0; // 设置为0，表示不删除任何记录

        // 2) 获取同步前目标表的记录数
        logger.info("正在获取目标表 {} 同步前的记录数...", tableName);
        int beforeCount = getTargetTableCount(tableName);
        logger.info("目标表 {} 同步前记录数: {} 条", tableName, beforeCount);

        // 3) 构建SELECT语句（不分页），依靠JDBC游标流式读取
        StringBuilder selectSql = new StringBuilder();
        selectSql.append("SELECT ");
        for (int i = 0; i < columns.size(); i++) {
            if (i > 0) selectSql.append(", ");
            selectSql.append("[").append(columns.get(i)).append("]");
        }
        selectSql.append(" FROM [").append(sourceTable).append("]");

        String upsertSql = buildUpsertSql(tableName, columns);


        try (Connection srcConn = DatabaseManager.getSqlserverConnection();
             PreparedStatement selectStmt = srcConn.prepareStatement(selectSql.toString(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
             Connection dstConn = DatabaseManager.getMysqlConnection();
             PreparedStatement upsertStmt = dstConn.prepareStatement(upsertSql)) {

            // SQLServer建议设置合理fetch size
            try {
                selectStmt.setFetchSize(100);
            } catch (SQLException ignore) { }

            // 批写需要手动控制事务，提升性能
            boolean originalAutoCommit = dstConn.getAutoCommit();
            dstConn.setAutoCommit(false);
            int currentBatch = 0;

            try (ResultSet rs = selectStmt.executeQuery()) {
                while (rs.next()) {
                    int paramIndex = 1;
                    for (String column : columns) {
                        Object value = rs.getObject(column);
                        upsertStmt.setObject(paramIndex++, value);
                    }
                    upsertStmt.addBatch();
                    currentBatch++;
                    stats.totalRows++;

                    // 每1000条记录输出一次进度日志
                    if (stats.totalRows % 1000 == 0) {
                        logger.info("UPSERT同步进度: 已处理 {} 条记录", stats.totalRows);
                    }

                    if (currentBatch >= batchSize) {
                        // 执行批量UPSERT
                        upsertStmt.executeBatch();
                        dstConn.commit();
                        currentBatch = 0;
                        result.setProcessedRows(stats.totalRows);
                        
            
                    }
                }

                if (currentBatch > 0) {
                    // 执行最后一批
                    upsertStmt.executeBatch();
                    dstConn.commit();
                    
        
                }
            } finally {
                dstConn.setAutoCommit(originalAutoCommit);
            }
        }
        
        // 4) 获取同步后目标表的记录数，计算准确的更新/插入数量
        logger.info("正在获取目标表 {} 同步后的记录数...", tableName);
        int afterCount = getTargetTableCount(tableName);
        logger.info("目标表 {} 同步后记录数: {} 条", tableName, afterCount);
        
        // 计算实际的更新和插入数量
        if (afterCount >= beforeCount) {
            // 如果记录数增加或不变，说明有插入操作
            int inserted = afterCount - beforeCount + stats.deletedRows;
            int updated = stats.totalRows - inserted;
            
            stats.insertedRows = Math.max(0, inserted);
            stats.updatedRows = Math.max(0, updated);
        } else {
            // 如果记录数减少，说明删除的记录比插入的多
            int netDeleted = beforeCount - afterCount;
            int inserted = stats.totalRows - netDeleted;
            
            stats.insertedRows = Math.max(0, inserted);
            stats.updatedRows = Math.max(0, stats.totalRows - inserted);
        }
        
        // 输出最终统计结果
        logger.info("UPSERT同步统计 - 总处理: {} 条, 插入: {} 条, 更新: {} 条, 删除: {} 条", 
            stats.totalRows, stats.insertedRows, stats.updatedRows, stats.deletedRows);
        
        return stats;
    }
    
    /**
     * 获取目标表的记录数
     */
    private int getTargetTableCount(String tableName) throws SQLException {
        try (Connection conn = DatabaseManager.getMysqlConnection();
             PreparedStatement stmt = conn.prepareStatement("SELECT COUNT(*) FROM `" + tableName + "`")) {
            try (ResultSet rs = stmt.executeQuery()) {
                if (rs.next()) {
                    return rs.getInt(1);
                }
            }
        }
        return 0;
    }

    /**
     * 删除目标表中源表没有的记录
     * 已禁用：直接使用INSERT，让数据库约束处理重复数据
     */
    /*
    private int deleteOrphanedRecords(String tableName) throws SQLException {
        int deletedRows = 0;
        
        try (Connection srcConn = DatabaseManager.getSqlserverConnection();
             Connection dstConn = DatabaseManager.getMysqlConnection()) {
            
            // 获取源表所有IMEI
            Set<String> sourceImeis = new HashSet<>();
            String selectImeiSql = "SELECT DISTINCT [" + keyField + "] FROM [" + sourceTable + "]";
    
            
            try (PreparedStatement stmt = srcConn.prepareStatement(selectImeiSql);
                 ResultSet rs = stmt.executeQuery()) {
                while (rs.next()) {
                    String imei = rs.getString(keyField);
                    if (imei != null && !imei.trim().isEmpty()) {
                        sourceImeis.add(imei);
                    }
                }
            }
            

            
            if (sourceImeis.isEmpty()) {
                logger.warn("源表 {} 中没有找到有效的 {} 值", sourceTable, keyField);
                return 0;
            }
            
            // 获取目标表所有IMEI
            // String targetKeyField = columnMapping.getOrDefault(keyField, keyField);
            // Set<String> targetImeis = new HashSet<>();
            // String selectTargetImeiSql = "SELECT DISTINCT `" + targetKeyField + "` FROM `" + tableName + "`";
            // logger.info("查询目标表IMEI的SQL: {}", selectTargetImeiSql);
            // try (PreparedStatement stmt = dstConn.prepareStatement(selectTargetImeiSql);
            //      ResultSet rs = stmt.executeQuery()) {
            //     while (rs.next()) {
            //         String imei = rs.getString(targetKeyField);
            //         if (imei != null && !imei.trim().isEmpty()) {
            //             targetImeis.add(imei);
            //     }
            // }
            // }
            
            // logger.info("目标表 {} 中找到 {} 个有效的 {} 值", tableName, targetImeis.size(), targetKeyField);
            // if (targetImeis.size() <= 10) {
            //     logger.info("目标表IMEI列表: {}", targetImeis);
            // } else {
            //     logger.info("目标表IMEI列表(前10个): {}", targetImeis.stream().limit(10).toList());
            // }
            
            // 找出目标表中有但源表中没有的IMEI
            // Set<String> orphanedImeis = new HashSet<>(targetImeis);
            // orphanedImeis.removeAll(sourceImeis);
            
            // logger.info("目标表中有但源表中没有的IMEI数量: {}", orphanedImeis.size());
            // if (orphanedImeis.size() <= 10) {
            //     logger.info("目标表中有但源表中没有的IMEI列表: {}", orphanedImeis);
            // } else {
            //     logger.info("目标表中有但源表中没有的IMEI列表(前10个): {}", orphanedImeis.stream().limit(10).toList());
            // }
            
            // if (orphanedImeis.isEmpty()) {
            //     logger.info("没有需要删除的记录");
            //     return 0;
            // }
            
            // 注意：删除逻辑已被禁用，以下代码不会执行
            // logger.info("注意：删除逻辑已被禁用，将保留目标表中的历史数据，不会删除 {} 条记录", orphanedImeis.size());
            // 删除逻辑已被禁用，不执行删除操作
            
        } catch (SQLException e) {
            logger.error("删除目标表多余记录失败", e);
            throw e;
        }
        
        // 删除逻辑已被禁用，始终返回0
        return 0;
    }
    */

    /**
     * 确保目标表关键字段具有唯一索引；如果不存在，先清理重复数据，再创建唯一索引
     */
    private void ensureTargetKeyUniqueness(String tableName) {
        String targetKeyField = columnMapping.getOrDefault(keyField, keyField);
        // 信息架构检查唯一索引是否存在（当前库）
        String checkSql = "SELECT COUNT(1) FROM information_schema.statistics " +
                "WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ? AND non_unique = 0";
        try (Connection conn = DatabaseManager.getMysqlConnection();
             PreparedStatement ps = conn.prepareStatement(checkSql)) {
            ps.setString(1, tableName);
            ps.setString(2, targetKeyField);
            try (ResultSet rs = ps.executeQuery()) {
                boolean hasUnique = false;
                if (rs.next()) {
                    hasUnique = rs.getInt(1) > 0;
                }
                if (!hasUnique) {
                    int removed = cleanupDuplicateKeys(tableName, targetKeyField);
                    String indexName = "uk_" + tableName + "_" + targetKeyField;
                    String alterSql = "ALTER TABLE `" + tableName + "` ADD UNIQUE KEY `" + indexName + "` (`" + targetKeyField + "` )";
                    try (PreparedStatement alter = conn.prepareStatement(alterSql)) {
                        alter.executeUpdate();
                    }
                }
            }
        } catch (SQLException e) {
            logger.error("确保目标表键列唯一索引失败", e);
        }
    }

    /**
     * 清理目标表中按键列的重复数据，保留systime最新的一条；若systime相同，按id较大的保留
     * 返回删除的行数
     */
    private int cleanupDuplicateKeys(String tableName, String targetKeyField) {
        // 需要存在列 `systime` 与 `id` 方可使用该清理策略；若缺失则退化为任意保留一条
        String deleteSql = "DELETE t1 FROM `" + tableName + "` t1 JOIN `" + tableName + "` t2 " +
                "ON t1.`" + targetKeyField + "` = t2.`" + targetKeyField + "` " +
                "AND ( (t1.`systime` < t2.`systime`) " +
                "OR (t1.`systime` = t2.`systime` AND IFNULL(t1.`id`,0) < IFNULL(t2.`id`,0)) )";
        try (Connection conn = DatabaseManager.getMysqlConnection();
             PreparedStatement ps = conn.prepareStatement(deleteSql)) {
            int affected = ps.executeUpdate();
            return affected;
        } catch (SQLException e) {
            // 退化方案：删除除最小rowid外的重复（依赖隐式行比较，MySQL无rowid，仅尽力而为）
            String fallback = "DELETE t1 FROM `" + tableName + "` t1 JOIN `" + tableName + "` t2 " +
                    "ON t1.`" + targetKeyField + "` = t2.`" + targetKeyField + "` AND t1.`id` < t2.`id`";
            try (Connection conn = DatabaseManager.getMysqlConnection();
                 PreparedStatement ps = conn.prepareStatement(fallback)) {
                return ps.executeUpdate();
            } catch (SQLException ex) {
                logger.error("退化清理重复失败，将无法创建唯一索引，后续UPSERT可能继续插入重复。", ex);
                return 0;
            }
        }
    }

    @Override
    public String getStrategyName() {
        return "UPSERT";
    }
    
    /**
     * UPSERT同步统计信息
     */
    private static class UpsertSyncStats {
        public int totalRows;
        public int updatedRows;
        public int insertedRows;
        public int deletedRows;
        
        public UpsertSyncStats() {
            this.totalRows = 0;
            this.updatedRows = 0;
            this.insertedRows = 0;
            this.deletedRows = 0;
        }
    }
}
